package org.spark.sample

import com.redis._
import spray.json._
import org.apache.spark.streaming.Seconds
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import spray.json.DefaultJsonProtocol
import com.redis.RedisClient
import scala.collection.mutable.ArrayBuffer

/**
 * @author 325175
 *  基于redis 保存中间结果，进行数量累计
 */
object StreamingWordCountAppWithRedis {

  def main(args: Array[String]) {

    var param = Array("local[2]", "10.118.58.17", "9999", "10.118.58.17", "6379", "streaming_world_count")

    if (args.length != 0) {
      param = args
    }

    val conf = new SparkConf().setAppName("StreamingWordCountAppWithRedis").setMaster(param(0))

    val sc = new SparkContext(conf) // 由于Streaming无法用于一般的RDD数据的接收，这边保留一个sc

    val ssc = new StreamingContext(sc, Seconds(10)) // 每10秒钟汇总一次数据

    val words = ssc.socketTextStream(param(1), param(2).toInt)

    val wordCounts = words.flatMap(x => x.split(" ")).map(x => (x, 1)).reduceByKey(_ + _)

    //准备要转换的代码
    case class Word(word: String, count: Int)
    // 这个是转换器
    object MyJsonProtocol extends DefaultJsonProtocol {
      implicit val wordFormat = jsonFormat2(Word)
    }

    import MyJsonProtocol._

    //Streaming的rdd不能直接使用，只能通过遍历的形式进行一个个关联
    //我们发现这样其实效率不是很高，后续会提供spark自身缓存的例子
    //wordCounts.print()

    wordCounts.foreachRDD(rdd => {

      // 读取已有的结果，进行转换,这里使用redis
      val r = new RedisClient(param(3), param(4).toInt)
      val savedJsons = r.get(param(5)) // 这里要json转对象

      println(savedJsons)
      // 这里使用模式匹配
      val saveResult = savedJsons match {
        case Some(s) => s.toString.parseJson.convertTo(DefaultJsonProtocol.listFormat[Word])
        case None    => List[Word]()
        case _       => List[Word]()
      }

      //转换成RDD
      val savedCounts = sc.parallelize(saveResult).map(v => (v.word, v.count))

      val totalCounts = rdd.union(savedCounts).reduceByKey(_ + _)

      // 保存结果到redis里面，先转换为List
      val resultList = ArrayBuffer[Word]()
      totalCounts.collect().foreach(r => resultList += Word(r._1, r._2))

      r.set(param(5), resultList.toArray.toJson)

    })

    ssc.start()
    ssc.awaitTermination()

  }
}